package org.momo.rpc.proxy;

import org.momo.rpc.circuitbreaker.CircuitBreakerProvider;
import org.momo.rpc.circuitbreaker.impl.CircuitBreaker;
import org.momo.rpc.client.Client;
import org.momo.rpc.client.impl.UnprocessedRequest;
import org.momo.rpc.config.RpcConfigLoader;
import org.momo.rpc.constant.Constant;
import org.momo.rpc.constant.LoadBalanceKey;
import org.momo.rpc.exception.RpcCallException;
import org.momo.rpc.extention.ExtensionLoader;
import org.momo.rpc.loadbalance.LoadBalanceProvider;
import org.momo.rpc.loadbalance.LoadBalancer;
import org.momo.rpc.protocol.RpcMessage;
import org.momo.rpc.protocol.RpcRequest;
import org.momo.rpc.protocol.RpcResponse;
import org.momo.rpc.protocol.constant.MessageStatus;
import org.momo.rpc.protocol.constant.ProtocolConstant;
import org.momo.rpc.provider.model.ProviderRegistryInfo;
import org.momo.rpc.registry.ConsumerRegistry;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;

public class SendRequestTask implements Callable<RpcMessage<RpcResponse>> {

    private final RpcMessage<RpcRequest> reqMessage;

    private final Class<?> methodReturnType;

    private final static Client client = ExtensionLoader.getExtensionLoader(Client.class).getExtension(RpcConfigLoader.getRpcConfig().getClient());

    private static final ConsumerRegistry consumerRegistry = ExtensionLoader.getExtensionLoader(ConsumerRegistry.class).getExtension(RpcConfigLoader.getRpcConfig().getRegistry());

    public SendRequestTask(RpcMessage<RpcRequest> reqMessage, Class<?> methodReturnType) {
        this.reqMessage = reqMessage;
        this.methodReturnType = methodReturnType;
    }

    @Override
    public RpcMessage<RpcResponse> call() throws Exception {
        RpcMessage.Header reqHeader = reqMessage.getHeader();
        RpcRequest request = reqMessage.getData();

        // fetch possible service
        List<ProviderRegistryInfo> possibleServiceList = consumerRegistry.discover(request.getServiceName(), request.getServiceVersion(), request.getServiceKey());
        if(possibleServiceList == null || possibleServiceList.isEmpty()) {
            throw new RpcCallException(reqHeader.getId(), request.getServiceName(), request.getMethodName(), true, "No available service!");
        }

        // load balance
        String loadBalanceName = String.join(Constant.LOADBALANCE_DELIMITER, request.getServiceName(), request.getServiceVersion(), request.getMethodName());
        LoadBalancer loadBalancer = LoadBalanceProvider.getLoadBalancer(loadBalanceName);
        if(loadBalancer == null) {
            // todo rpc config
            LoadBalanceProvider.initLoadBalancer(loadBalanceName, LoadBalanceKey.POLLING);
            loadBalancer = LoadBalanceProvider.getLoadBalancer(loadBalanceName);
        }
        ProviderRegistryInfo service = loadBalancer.loadBalance(request, possibleServiceList);

        // circuit breaker not allow this service
        String serviceAddress = String.join(Constant.SERVICE_DELIMITER, request.getServiceName(), request.getServiceVersion());
        CircuitBreaker circuitBreaker = CircuitBreakerProvider.getCircuitBreaker(serviceAddress);
        if(!circuitBreaker.allowRequest()) {
            throw new RpcCallException(reqHeader.getId(), request.getServiceName(), request.getMethodName(), false, "Circuit not allow the request, refuse it!");
        }

        // get ticket for response
        InetSocketAddress serverAddress = new InetSocketAddress(service.getHost(), service.getPort());
        CompletableFuture<RpcMessage<RpcResponse>> resultFuture = UnprocessedRequest.put(reqMessage.getHeader().getId(), reqMessage, circuitBreaker);
        client.sendRequest(reqMessage, serverAddress);

        // 阻塞等待 resMessage
        RpcMessage<RpcResponse> resMessage = resultFuture.get();

        // resMessage 错误, 抛出 RpcCallException
        // 1. 响应为空
        if (resMessage == null || resMessage.getHeader() == null || resMessage.getData() == null || resMessage.getData().getData() == null) {
            circuitBreaker.recordFailure();
            throw new RpcCallException(reqHeader.getId(), request.getServiceName(), request.getMethodName(), true, "Response error");
        }
        // 2. 魔数
        if (resMessage.getHeader().getMagic() != ProtocolConstant.MAGIC) {
            throw new RpcCallException(reqHeader.getId(), request.getServiceName(), request.getMethodName(), true, "Magic error");
        }
        // 3. 响应状态错误
        if (resMessage.getHeader().getStatus() == MessageStatus.RESPONSE_ERROR.getCode()) {
            circuitBreaker.recordFailure();
            throw new RpcCallException(reqHeader.getId(), request.getServiceName(), request.getMethodName(), true, "Response error");
        }
        // 4. 返回数据类型
        if (!resMessage.getData().getDataType().equals(this.methodReturnType)) {
            circuitBreaker.recordFailure();
            throw new RpcCallException(reqHeader.getId(), request.getServiceName(), request.getMethodName(), true, "Response error");
        }
        return resMessage;
    }

}
